package com.tca.common.learning.webflux.reactor.start;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/**
 * 特点
 * 1.相对于传统的基于回调和Future的异步开发方式, 响应式编程更加具有可编排性和可读性, 配合lambda表达式, 代码更加简洁,
 * 处理逻辑的表达就像装配"流水线", 适用于对数据流的处理
 * 2.在订阅(subscribe)时才触发数据流, 这种数据流叫做"冷"数据流, 就像插座插上电器才会有电流一样, 还有一种数据流不管是
 * 否有订阅者订阅它都会一直发出数据, 称之为"热"数据流, Reactor中几乎都是"冷"数据流
 * 3.调度器对线程管理进行更高层次的抽象, 使得我们可以非常容易地切换线程执行环境
 * 4.灵活的错误处理机制有利于编写健壮的程序
 * 5."回压"机制使得订阅者可以无限接受数据并让它的源头"满负荷"推送所有的数据, 也可以通过使用request方法来告知源头它一次
 * 最多能够处理 n 个元素, 从而将"推送"模式转换为"推送+拉取"混合的模式
 */
@Slf4j
public class FluxAndMonoTest {

    public static void main(String[] args) {

        // 创建流
//        buildStream();

        // 创建并订阅
//        buildAndSubscribe();

        // 验证
//        verifier();

        // map操作符 + 验证
//        mapAndVerifier();

        // flatMap操作符
//        flatMap();

        // filter操作符
        filter();

    }



    /**
     * 创建流
     */
    private static void buildStream() {
        log.info("创建流");
        // 1.Flux 对象实现发布者, 返回N个元素; Mono 也实现发布者, 返回0或1个元素
        Flux.just(1, 2, 3, 4, 5, 6);
        Mono.just(1);
        // 2.根据数组和集合创建流
        Integer[] intArray = new Integer[]{1, 2, 3, 4, 5, 6};
        Flux.fromArray(intArray);
        List<Integer> list = Lists.newArrayList(1, 2, 3, 4, 5, 6);
        Flux.fromIterable(list);
        Flux.fromStream(list.stream());
        Flux.fromStream(Arrays.stream(intArray));
        // 3.创建空数据流
        Flux.just();
        Flux.empty();
        Mono.empty();
        Mono.justOrEmpty(Optional.empty());
        // 4.只有错误信号的数据流
        Flux.error(new Exception());
        Mono.error(new Exception());
    }

    /**
     * 创建流并订阅
     */
    private static void buildAndSubscribe() {
        log.info("创建流并订阅");
        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::println);
        // subscribe重载
        Flux.just(1, 2, 3).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("completed!")
        );
    }

    /**
     * verifier
     * 使用verifier, 如果校验失败会抛出异常
     */
    private static void verifier() {
        StepVerifier.create(Mono.just(1))
                .expectNext(2)
                .expectComplete()
                .verify();
    }


    /**
     * map & verifier
     */
    private static void mapAndVerifier() {
        StepVerifier.create(Flux.range(1, 6)
                        .map(i -> i * i))
                .expectNext(1, 4, 9, 16, 25, 36)
                .expectComplete()
                .verify();

    }

    /**
     * flatmap: 操作可以将每个数据元素转换/映射为一个流，然后将这些流合并为一个大的数据流
     */
    private static void flatMap() {
        Flux.just("a", "b", "c")
                .flatMap(s -> Flux.fromArray(s.split("\\s*"))
                        .doOnNext(System.out::println));

    }

    /**
     * filter
     */
    private static void filter() {
        StepVerifier.create(Flux.range(1, 6)
                        .filter(i -> i % 2 == 1)
                        .map(i -> i * i))
                .expectNext(1, 9, 25)
                .verifyComplete();
    }


}
